package org.jetlinks.pro.things.impl.rpc;

import lombok.AllArgsConstructor;
import org.jetlinks.core.enums.ErrorCode;
import org.jetlinks.core.message.RepayableThingMessage;
import org.jetlinks.core.message.ThingMessage;
import org.jetlinks.core.message.function.FunctionInvokeMessage;
import org.jetlinks.core.message.property.ReadPropertyMessage;
import org.jetlinks.core.message.property.WritePropertyMessage;
import org.jetlinks.core.things.*;
import org.jetlinks.pro.things.ThingsConnector;
import org.jetlinks.pro.things.ThingsConnectorManager;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.List;
import java.util.Map;
import java.util.function.Function;


/**
 * 连接到物的RPC支持，用于将请求消息转换为对应物的连接器支持
 *
 * @author zhouhao
 * @since 1.12
 */
@AllArgsConstructor
public class ConnectedThingRpcSupport implements ThingRpcSupport {

    //物实例接口，可通过此接口获取物的物模型以及相关配置信息
    private final Thing thing;

    //物连接器管理器,用于管理连接器,并从连接器中获取数据进行持久化等处理.
    private final ThingsConnectorManager connectorManager;

    /**
     * 将请求消息转换为对应物的连接器支持
     *
     * @param message 请求消息
     * @return 物连接器
     */
    private Mono<ThingsConnector> getConnector(ThingMessage message) {
        //调用功能
        if (message instanceof FunctionInvokeMessage) {
            String function = ((FunctionInvokeMessage) message).getFunctionId();
            return connectorManager
                .getConnector(ThingId.of(thing.getType().getId(), thing.getId()),
                              MetadataId.of(ThingMetadataType.function, function));

        }
        //读取属性
        if (message instanceof ReadPropertyMessage) {
            List<String> properties = ((ReadPropertyMessage) message).getProperties();
            if (StringUtils.isEmpty(properties) || properties.size() > 1) {
                return Mono.empty();
            }
            return connectorManager
                .getConnector(ThingId.of(thing.getType().getId(), thing.getId()),
                              MetadataId.of(ThingMetadataType.property, properties.get(0)));

        }
        //修改属性
        if (message instanceof WritePropertyMessage) {
            Map<String, Object> properties = ((WritePropertyMessage) message).getProperties();
            if (StringUtils.isEmpty(properties) || properties.size() > 1) {
                return Mono.empty();
            }
            String property = properties.keySet().iterator().next();
            return connectorManager
                .getConnector(ThingId.of(thing.getType().getId(), thing.getId()),
                              MetadataId.of(ThingMetadataType.property, property));

        }

        return Mono.empty();
    }

    /**
     * RPC调用
     *
     * @param message RPC请求消息
     * @return 多个RPC的调用结果
     */
    @Override
    public Flux<? extends ThingMessage> call(ThingMessage message) {
        return this
            .getConnector(message)
            //过滤掉不支持RPC的连接器
            .filter(connector -> connector.isWrapperFor(ThingRpcSupport.class))
            .map(connector -> connector
                .unwrap(ThingRpcSupport.class)
                .call(message)
                .cast(ThingMessage.class))
            .switchIfEmpty(onConnectorUnsupported(message))
            .flatMapMany(Function.identity());
    }

    /**
     * 当没有支持的RPC连接器时
     *
     * @param message RPC请求消息
     * @return 异常类型的物消息
     */
    private Mono<Flux<ThingMessage>> onConnectorUnsupported(ThingMessage message) {
        if (message instanceof RepayableThingMessage) {
            return Mono.just(Flux.just(((RepayableThingMessage<?>) message)
                                           .newReply()
                                           .error(ErrorCode.UNSUPPORTED_MESSAGE)));
        }
        return Mono.error(new UnsupportedOperationException());
    }
}
